package org.jpush.core;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.jpush.core.AbstractChannelManager.ChannelContext;
import org.jpush.core.AbstractChannelStreamer.SendResult;
import org.jpush.core.Protocol.Result;
import org.jpush.core.socket.JPushClientSocket;
import org.jpush.packages.OutgoingPacket;
import org.jpush.utils.DataBuffer;
import org.jpush.utils.Pointer;

public class PhysicalConnection
{
	public static enum ConnectionStatus
	{
	    Disposable,
	    WaitingForWrite,
	    Connected,       // Means first time to communicate
	    Attached         // Means connection has been created.(After Connected)
	};
	
	private Server server;
	private ChannelContext channelCxt;
	private ConnectionContext connectionContext;

	private ListenerOptions listenerOption;
	private ConnectionStatus status;
	
	// Indicate how many worker threads are working on this connection 
	private AtomicInteger ioWorkersReferenceCounter;
	
	// When this connection is created
	private long creationTime;
	
	
	private Lock csLock;

	// If it is a connection of monitor
	private boolean isObserver;

	// Used by connection of monitor only.
	private String clientKey;

	// Used by connection of client only.
	private LogicalConnection logicalConnection;

	private DataBuffer inBuffer;
	private DataBuffer outBuffer;
	private boolean writeInProgress;

	/* Saved socket information */
	private JPushClientSocket socket;
	private int peerPort;
	private String peerIP;
	
	public PhysicalConnection(JPushClientSocket socket, 
			boolean blnIsObserver, ListenerOptions listenerOption, 
			Server server)
	{
		this.socket = socket;
		
		// Save the client's IP and port
		resetPeer(socket);

		this.creationTime = System.currentTimeMillis();
		this.writeInProgress = false;
	    this.isObserver = blnIsObserver;
	    
	    this.status = ConnectionStatus.Connected;
	    this.listenerOption = listenerOption;

	    this.ioWorkersReferenceCounter = new AtomicInteger(0);

	    this.outBuffer = new DataBuffer(listenerOption.getIntermediateSendBufferSize());
	    this.inBuffer = new DataBuffer(listenerOption.getIntermediateReceiveBufferSize());

	    this.channelCxt = new ChannelContext();

	    this.connectionContext = null;

	    this.csLock = new ReentrantLock();
	    
	    this.server = server;
	}
	
	/**
	 * Check if state represents connected
	 * @param state
	 * @return
	 */
	public static boolean notConnected(ConnectionStatus state)
	{
		return (state == ConnectionStatus.Disposable || state == ConnectionStatus.WaitingForWrite);
	}
	
	/**
	 * Save the IP and port of the remote client
	 * 
	 * @param socket
	 */
	private void resetPeer(JPushClientSocket socket)
	{
    	SocketAddress sockAddr = socket.getRemoteAddress();
    	
        if ((sockAddr != null) && (sockAddr instanceof InetSocketAddress))
        {
        	InetSocketAddress inetSockAddr = (InetSocketAddress)sockAddr;
        	this.peerIP = inetSockAddr.getAddress().getHostAddress();
        	this.peerPort = inetSockAddr.getPort();
        }
	}


	public ConnectionStatus getStatus()
	{
		return this.status;
	}

	public Protocol getProtocol()
	{
	    return listenerOption.getProtocol();
	}

	public boolean isObserverChannel()
	{
	    return isObserver;
	}

	public String getLogicalConnectionKey()
	{
	    if (isObserverChannel())
	    {
	        return clientKey;
	    }
	    else
	    {
	        return logicalConnection.getKey();
	    }
	}

	public LogicalConnection getLogicalConnection()
	{
	    return logicalConnection;
	}

	public boolean checkIfUnusedByIOWorkers()
	{
	    return status == ConnectionStatus.Disposable && 
	    	ioWorkersReferenceCounter.intValue() == 0;
	}

	public int getIoWorkersReferenceCounter()
	{
	    return ioWorkersReferenceCounter.intValue();
	}

	public void incrementIoWorkersReferenceCounter()
	{
		ioWorkersReferenceCounter.incrementAndGet();
	}

	public void decrementIoWorkersReferenceCounter()
	{
		ioWorkersReferenceCounter.decrementAndGet();
	}

	public int getPeerPort()
	{
	    return peerPort;
	}

	public String getPeerIP()
	{
	    return peerIP;
	}

	public double getLifeDuration()
	{
		return System.currentTimeMillis() - creationTime;
	}

	public void close(boolean bWaitForSendsToComplete)
	{
		csLock.lock();
		
	    if (status.ordinal() < ConnectionStatus.Connected.ordinal())
	    {
			csLock.unlock();
	        return;
	    }

	    //Either connected or attached
	    if (bWaitForSendsToComplete)
	    {
	        if (outBuffer.getDataSize() == 0)
	        {
	            closeSocket();
	        }
	        else
	        {
	            status = ConnectionStatus.WaitingForWrite;
	        }
	    }
	    else
	    {
	        closeSocket();
	    }
		csLock.unlock();
    }

	public void postReceive()
	{
	    if (PhysicalConnection.notConnected(status))
	    {
	        return;
	    }
	    
	    //socket.read(m_wsaInBuffer, null, serverImpl.getDispatcher().getPhysicalConnectionReadHandler());
    	
//	    try
//	    {
//	    	socket.read(m_wsaInBuffer);
//	    }
//	    catch (IOException ioe)
//	    {
//	    	ioe.printStackTrace();
//	    	return;
//	    }
	    incrementIoWorkersReferenceCounter();
	}

	public ConnectionStatus readReceivedBytes()
	{
	    if(PhysicalConnection.notConnected(status))
	    {
	        return status;
	    }

	    //inBuffer.append(m_wsaInBuffer.array());
	    try
	    {
			socket.read(inBuffer);
		}
	    catch (IOException e)
	    {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

	    return status;
	}

	public SendResult pushPacket(OutgoingPacket packet)
	{
		csLock.lock();
		SendResult result = pushPacketCommon(packet);
		csLock.unlock();
		
		return result;
	}

	private SendResult pushPacketCommon(OutgoingPacket packet)
	{
	    if (PhysicalConnection.notConnected(status))
	    {
	        return SendResult.SendResult_NotOK;
	    }

	    int iWrittenBytes = 0;

	    // Use the protocol the convert the packet into bytes
	    Protocol protocol = listenerOption.getProtocol();
	    Result iResult = protocol.serializeOutgoingPacket(packet, outBuffer, iWrittenBytes);

	    if (iResult != Result.Success)
	    {
	    	server.getServerStats().addToCumul(ServerStats.Measures.BandwidthRejection, iWrittenBytes);
	        return iResult == Result.eInsufficientBuffer ? 
	        		SendResult.SendResult_Retry : SendResult.SendResult_NotOK;
	    }

	    server.getServerStats().addToCumul(ServerStats.Measures.BandwidthOutstanding, iWrittenBytes);
	    Service service = server.getDispatcher().getCurrentService();
	    if (service != null)
	    {
	    	server.getServerStats().addToDistribution(
	    			ServerStats.Measures.BandwidthOutboundVolPerRequest, service.getName(), iWrittenBytes);
	    }

	    if (!writeInProgress)
	    {
	        return writeBytes() ? SendResult.SendResult_OK : SendResult.SendResult_NotOK;
	    }

	    return SendResult.SendResult_OK;
	}

	public SendResult tryPushPacket(OutgoingPacket pPacket)
	{
		if(csLock.tryLock())
		{
			SendResult result = pushPacketCommon(pPacket);
			csLock.unlock();
			return result;
		}
	    return SendResult.SendResult_Retry;
	}

	/**
	 * Called when the packet is sent successfully
	 *
	 */
	public ConnectionStatus onSendCompleted(int dwIoSize, Pointer<Boolean> pBIsBufferIdle)
	{
		csLock.lock();
		if(status.ordinal() < ConnectionStatus.WaitingForWrite.ordinal())
		{
			csLock.unlock();
		    return status;
		}
	
	    outBuffer.pop(dwIoSize);

	    // Check if there is no data in the buffer
	    boolean bIsBufferIdle = outBuffer.getDataSize() == 0;
	    pBIsBufferIdle.setValue(Boolean.valueOf(bIsBufferIdle));

	    if (bIsBufferIdle)
	    {
	        if (status == ConnectionStatus.WaitingForWrite)
	        {
	            closeSocket();
	        }
	        else
	        {
	            writeInProgress = false;
	        }
			csLock.unlock();
	        return status;
	    }

	    writeBytes();
		csLock.unlock();
	    return status;
	}


	public DataBuffer getReceiveBuffer()
	{
	    return inBuffer;
	}

	private boolean writeBytes()
	{		
//		int dwToPick = Math.min(listenerOption.sendBufferSize, oBuffer.getDataSize());
//		m_wsaOutBuffer.put(oBuffer.getBuffer(), m_wsaOutBuffer.arrayOffset(), dwToPick);
		
//		try
//		{
//			//socket.configureBlocking(false);
//			socket.write(m_wsaOutBuffer);
//		}
//		catch (IOException ioe)
//		{
//			ioe.printStackTrace();
//			return false;
//		}
		
		try
		{
			socket.write(outBuffer);
		}
		catch (IOException e)
		{
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
		//socket.write(m_wsaOutBuffer, null, serverImpl.getDispatcher().getPhysicalConnectionWriteHandler());

	    writeInProgress = true;
	    incrementIoWorkersReferenceCounter();
	    return true;
	}


	private void closeSocket()
	{
		socket.close();
	    status = ConnectionStatus.Disposable;
	}

	public void setConnectionContext(ConnectionContext pConnectionContext)
	{
	    this.connectionContext = pConnectionContext;
	}

	public ConnectionContext getConnectionContext()
	{
	    return connectionContext;
	}

	public void attachToClient(LogicalConnection logicalConnection)
	{
	    this.logicalConnection = logicalConnection;
	    this.status = ConnectionStatus.Attached;
	}

	public void attachToClient(String clientKey)
	{
	    this.clientKey = clientKey;
	    this.status = ConnectionStatus.Attached;
	}

//	public Socket getSocket()
//	{
//	    return socket.socket();
//	}
	
	public ChannelContext getChannelContext()
	{
		return this.channelCxt;
	}
}
